package momo.app.multicast.presence;

import java.util.Vector;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

public class Presence {
	private Log log = LogFactory.getLog(Presence.class);
	
	private enum PresenceState {
		IN_PRESENCE, DELAY_ACKNOWLEDGEMENT, FREE
	}
	
//	private static double AVERAGE_RTT_CONSTANT = 0.9;
	private static long PRESENCE_INTERVAL = 10 * 1000;
	private static long INITIAL_RTT = 5000;

	private long round_trip_time;
	private PresenceState state;

	private Vector<PresenceMessage> return_value_buffer;
	private boolean wait_for_rtt_timeout;
	private Boolean acknowledgement_requested;
//	private long last_presence_initiate_time;

	private PresenceMulticastComm connection;
	private PresenceCallback callback;
	
	 private ScheduledExecutorService initiate_presence_scheduler;
	 private ScheduledFuture<?> taskPresence;
	 private ScheduledExecutorService rtt_timeout_scheduler;
	 private ScheduledFuture<?> taskRtt;
	 private ScheduledExecutorService after_presence_scheduler;
	 private ScheduledFuture<?> taskAfterPresence;


//	public static void setAverageRTTConstant(double new_constant) {
//		AVERAGE_RTT_CONSTANT = new_constant;
//	}
//
	public static void setPresenceInterval(long new_interval) {
		PRESENCE_INTERVAL = new_interval;
	}

	public static void setInitialRTT(long new_initial_rtt) {
		INITIAL_RTT = new_initial_rtt;
	}

	/**
	 * 
	 * @param interval
	 *            Waiting time before a client initiate presence.
	 * @param node
	 * @param topicID
	 * @param callback
	 */
	public Presence(PresenceMulticastComm connection, PresenceCallback callback) {
		this.connection = connection;
		this.callback = callback;
		this.state = PresenceState.FREE;
		this.acknowledgement_requested = false;
		round_trip_time = INITIAL_RTT;
		wait_for_rtt_timeout = false;
		
		rtt_timeout_scheduler = Executors.newScheduledThreadPool(1);
		initiate_presence_scheduler = Executors.newScheduledThreadPool(1);
		after_presence_scheduler = Executors.newScheduledThreadPool(1);
//		taskPresence = initiate_presence_scheduler.schedule(createInitiatePresenceTask(), PRESENCE_INTERVAL, TimeUnit.MILLISECONDS);
	}

	public void finalize() {
		if (taskPresence != null)
			taskPresence.cancel(false);
		if (taskRtt != null)
			taskRtt.cancel(false);
		if (taskAfterPresence != null)
			taskAfterPresence.cancel(false);
	}

	private Runnable createInitiatePresenceTask() {
		return new Runnable()
		{
			public void run()
			{
				initiatePresence();
			}
		};
	}
	
	private Runnable createRttTimeoutTask() {
		return new Runnable() {
			public void run() {
				onRttTimeout();
			}
		};
	}

	private Runnable createAfterPresenceTask() {
		return new Runnable() {
			public void run() {
				onAfterPresenceTimeout();
			}
		};
	}

	private synchronized void onAfterPresenceTimeout() {
		taskAfterPresence = after_presence_scheduler.schedule(
				createInitiatePresenceTask(), PRESENCE_INTERVAL, TimeUnit.MILLISECONDS);
		setState(PresenceState.FREE);
		if (this.acknowledgement_requested.booleanValue()) {
			handleNewPeerJoin();
		}
	}
	
	/**
	 * Initiate presence.
	 */
	protected synchronized void initiatePresence() {
		log.warn("initiate presence " + this);
		setState(PresenceState.IN_PRESENCE);
		return_value_buffer = new Vector<PresenceMessage>();
		wait_for_rtt_timeout = true;
		if (taskRtt != null)
			taskRtt.cancel(false);
		if (taskAfterPresence != null)
			taskAfterPresence.cancel(false);
		if (taskPresence != null)
			taskPresence.cancel(false);
		taskRtt = rtt_timeout_scheduler.schedule(createRttTimeoutTask(), 4 * round_trip_time, TimeUnit.MILLISECONDS);
//		last_presence_initiate_time = System.currentTimeMillis();
		callback.onInitiatePresence();
		sendPresenceMessage();
	}

	/**
	 * Called when a presence message is received.
	 * 
	 * @param message_content
	 */
	public synchronized void onReceivePresenceMessage(
			PresenceMessage message_content) {
		log.warn("on receive presence message");
		if (wait_for_rtt_timeout) {
			log.warn("still in presence session");
			return_value_buffer.add(message_content);
		} else {
			log.warn("other initiate presence");
			setState(PresenceState.IN_PRESENCE);
			// Other peer initiate presence
			if (taskPresence != null) {
				taskPresence.cancel(false);
			}
			if (taskAfterPresence != null) {
				taskAfterPresence.cancel(false);
			}
			//initiate_presence_timer.purge();
			taskRtt = rtt_timeout_scheduler.schedule(createRttTimeoutTask(),
					4 * round_trip_time, TimeUnit.MILLISECONDS);
//			last_presence_initiate_time = System.currentTimeMillis();
			wait_for_rtt_timeout = true;
			return_value_buffer = new Vector<PresenceMessage>();
			return_value_buffer.add(message_content);
			callback.onInitiatePresence();
			sendPresenceMessage();
		}
	}

	/**
	 * Called after 4 times average round trip time after starting of first
	 * presence message is received.
	 */
	protected synchronized void onRttTimeout() {
		System.out.println("rtt timeout");
		setState(PresenceState.DELAY_ACKNOWLEDGEMENT);
		connection.setAcknowledgementReceived(false); //connection.setAcknowledgementReceived(false);
//		updateRoundTripTime(System.currentTimeMillis()
//				- last_presence_initiate_time);
		wait_for_rtt_timeout = false;
		if (taskPresence != null) {
			taskPresence.cancel(false);
			//initiate_presence_timer.purge();
		}
		taskAfterPresence = after_presence_scheduler.schedule(createAfterPresenceTask(),
				2 * round_trip_time, TimeUnit.MILLISECONDS);
		Vector<PresenceMessage> return_value = return_value_buffer;
		return_value_buffer = null;
		connection.setNetworkState(NetworkState.CONNECTED);
		this.callback.onReceivePresenceResult(return_value);
	}

	private void sendPresenceMessage() {
		
		PresenceMessage message = new PresenceMessage(callback.getUserName(),
				callback.getVersion(), callback.getVectorClock());
		
		connection.publishPresenceMessage(message);
		
	}

//	/**
//	 * Calculate new average round trip time.
//	 * 
//	 * @param new_time
//	 *            Last round trip time.
//	 */
//	private void updateRoundTripTime(long new_time) {
//		round_trip_time = (long) (AVERAGE_RTT_CONSTANT * round_trip_time + (1 - AVERAGE_RTT_CONSTANT)
//				* new_time);
//	}

	public void resetRTTTimeout() {
		log.warn("reset rtt timeout");
		round_trip_time = INITIAL_RTT;
	}

	public synchronized void cancelPresence() {
		log.warn("cancel presence");
		setState(PresenceState.FREE);
		if (taskRtt != null) {
			taskRtt.cancel(false);
			//rtt_timeout_timer.purge();
		}
		wait_for_rtt_timeout = false;
		if (taskPresence != null) {
			taskPresence.cancel(false);
			//initiate_presence_timer.purge();
		}
		if (taskAfterPresence != null) {
			taskAfterPresence.cancel(false);
		}
//		taskPresence = initiate_presence_scheduler.schedule(createInitiatePresenceTask(),
//				PRESENCE_INTERVAL, TimeUnit.MILLISECONDS);
		return_value_buffer = new Vector<PresenceMessage>();
//		log.warn("end of cancel presence");
	}

	public void handleNewPeerJoin() {
		log.warn("handle new peer join");
		if (getState().equals(PresenceState.FREE)) {
			connection.publishAcknowledgement();				
			initiatePresence();
		}
		synchronized(this.acknowledgement_requested) {
			this.acknowledgement_requested = this.getState() != PresenceState.FREE;
		}
//		log.warn("end of handle new peer join");
	}
	
	private void setState(PresenceState state) {
		synchronized(this.state) {
			log.warn("presence state: " + state.toString());
			this.state = state;
		}
	}
	
	private PresenceState getState() {
		synchronized(this.state) {
			return this.state;
		}
	}
}
